package com.hbwxz.rabbitmq;

import com.hbwxz.constant.Constants;
import com.hbwxz.utils.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 备份交换机应用场景：如果我不想设置 mandatory 为 true，也想保证消息不丢失怎么办；这个时候备份交换机就出现了
 * <p>
 * 演示这个的时候需要每次删除 交换机 和 队列，不然会报错
 * <p>
 * Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method:
 * #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'alternate-exchange' for exchange 'myExchange' in vhost '/':
 * received the value 'Constants.AE_EXCHANGE' of type 'longstr' but current is none, class-id=40, method-id=10)
 *
 * @author shenzw
 * @date 2023/11/9
 */
@Slf4j
public class RabbitMqFaceTwo {
    public static void main(String[] args) throws IOException, TimeoutException {

        // 获取 channel
        Channel channel = RabbitMqUtils.getChannel();

        // 小 tips：fanout 类型交换机不需要指定 routingKey，因为 fanout 类型的交换机会将消息发送与所有与之相关的队列上
        /// =========================================================== 创建备份交换机🐔 ==================================================================================== ///
        channel.exchangeDeclare(Constants.AE_EXCHANGE, Constants.FANOUT_EXCHANGE, true); // 交换机的名字，交换机的类型，是否持久化
        channel.queueDeclare(Constants.AE_QUEUE, true, false, false, null);
        channel.queueBind(Constants.AE_QUEUE, Constants.AE_EXCHANGE, Constants.AE_ROUTING_KEY); // 队列，交换机，routingKey

        // 需要设置交换机的参数
        Map<String, Object> argsM = new HashMap<>(10);
        argsM.put("alternate-exchange", Constants.AE_EXCHANGE);
        channel.exchangeDeclare(Constants.MY_EXCHANGE, Constants.DIRECT_EXCHANGE, true, false, argsM); // 交换机的名字，交换机的类型，是否持久化，不知道，参数
        channel.queueDeclare(Constants.MY_QUEUE, true, false, false, null);
        channel.queueBind(Constants.MY_QUEUE, Constants.MY_EXCHANGE, Constants.MY_ROUTING_KEY); // 队列，交换机，routingKey

        // 我的生产者
        myProducer(channel);

        // 我的消费者
        myConsumer(channel);

        // 关闭连接
        RabbitMqUtils.close(channel);

    }

    /**
     * 我的消费者
     *
     * @param channel 信道
     */
    private static void myConsumer(Channel channel) throws IOException {
        // 消费消息（队列，是否自动确认消费消息【关闭，不然消息没有正常消费的情况下也会自动确认】，交换机，消费者【具体执行消费的相关逻辑】）
        channel.basicConsume(Constants.MY_QUEUE, false, Constants.MY_EXCHANGE, new DefaultConsumer(channel) {
            /**
             * 消费消息的核心方法，覆盖父类方法
             * @param consumerTag 消息者 ID
             * @param envelope Envelope（存放消息唯一 ID 等一些相关信息）
             * @param properties AMQP.BasicProperties
             * @param body 消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                long deliveryTag = envelope.getDeliveryTag(); // 唯一的 ID
                log.info("MyConsumer handleDelivery receive message = {}，deliveryTag = {}", new String(body), deliveryTag);
                System.out.println("MyConsumer handleDelivery receive message = " + new String(body) + "，deliveryTag = " + deliveryTag);
                channel.basicAck(deliveryTag, true); // 手动 Ack
            }
        });
    }

    /**
     * 我的生产者
     *
     * @param channel 信道
     */
    private static void myProducer(Channel channel) throws IOException {

        byte[] msg = "Hello RabbitMQ".getBytes();

        // 交换机名称，路由 KEY，mandatory， AMQP.BasicProperties()，发送消息
        // 模拟演示“我的交换机”失败路由到“我的队列”，然后消息路由到备份交换机中
        channel.basicPublish(Constants.MY_EXCHANGE, Constants.MY_ERROR_ROUTING_KEY, false,
                new AMQP.BasicProperties().builder()
                        .deliveryMode(2) // 消息持久化设置
                        .contentType("text/plain")
                        .build(),
                msg);
    }
}
